package com.ysy.mq;


import com.alibaba.fastjson2.JSON;
import com.github.phantomthief.collection.BufferTrigger;
import com.ysy.constant.RedisConstant;
import com.ysy.constant.RocketMqConstant;
import com.ysy.entity.UserCount;
import com.ysy.entity.dto.FollowCancelCountDTO;
import com.ysy.enums.FollowCancelTypeEnum;
import com.ysy.mapper.UserCountMapper;
import com.ysy.persimmons.common.util.StringUtils;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

@Component
@RocketMQMessageListener(consumerGroup = "persimmons_count_group_"+RocketMqConstant.FANS_COUNT_TOPIC,
        topic = RocketMqConstant.FANS_COUNT_TOPIC)
public class FansCountConsumer implements RocketMQListener<String> {

    private static Logger log = LoggerFactory.getLogger(FansCountConsumer.class);


    @Resource
    private UserCountMapper userCountMapper;

    @Resource
    private RedisTemplate<String,Object> redisTemplate;

    @Resource
    private RocketMQTemplate rocketMQTemplate;


    private BufferTrigger<String> bufferTrigger = BufferTrigger.<String>batchBlocking()
            .bufferSize(10000)
            .batchSize(1000)
            .linger(Duration.ofSeconds(1))
            .setConsumerEx(this::bufferTriggerConsume)
            .build();



    @Override
    public void onMessage(String message) {
        // 判断类型是关注还是取消
        if (StringUtils.isBlank(message)) {
            log.warn("当前mq消费数据库为空，暂不处理");
            return;
        }
        // 使用聚合消息
        bufferTrigger.enqueue(message);
    }

    private void bufferTriggerConsume(List<String> messages) {
        log.info("==> 聚合消息数量: {}", messages.size());

        // 处理消息
        List<FollowCancelCountDTO> dtoList = messages.stream()
                .map(message -> JSON.parseObject(message, FollowCancelCountDTO.class))
                .toList();

        // 分组
        Map<Long, List<FollowCancelCountDTO>> collectMap = dtoList.stream()
                .collect(Collectors.groupingBy(FollowCancelCountDTO::getUserId));


        Map<Long, Integer> mapCount = new HashMap<>();
        collectMap.forEach((userId, groupList) -> {
            int currentCount = 0;
            for (FollowCancelCountDTO dto : groupList) {
                FollowCancelTypeEnum followCancelTypeEnum = FollowCancelTypeEnum.get(dto.getType());
                if (Objects.isNull(followCancelTypeEnum)) {
                    log.info("{}聚合消息处理，未知的类型：{}", userId,dto.getType());
                    continue;
                }
                switch (followCancelTypeEnum){
                    case FOLLOW -> currentCount = currentCount + 1;
                    case CANCEL -> currentCount = currentCount - 1;
                }
            }
            mapCount.put(userId,currentCount);
        });

        // 存储redis 和数据库
        mapCount.forEach((userId, count) -> {
           String redisUserKey = RedisConstant.getUserCountKey(userId);
            Boolean b = redisTemplate.hasKey(redisUserKey);
            //  不存在 就去插入数据库兜底，记得存根
            if (b) {
                redisTemplate.opsForHash().increment(redisUserKey, RedisConstant.HASH_FIELD_FANS_TOTAL, count);
            }
        });


        // 发送 MQ, 计数数据落库
        // 构建消息体 DTO
        Message<String> messageDB = MessageBuilder.withPayload(JSON.toJSONString(mapCount))
                .build();

        // 异步发送 MQ 消息
        rocketMQTemplate.asyncSend(RocketMqConstant.FANS_COUNT_DB_TOPIC, messageDB, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("粉丝数入库MQ发送成功SendResult: {}", sendResult);
            }
            @Override
            public void onException(Throwable throwable) {
                log.error("计数粉丝数入库MQ发送异常: ", throwable);
            }
        });



    }
}
